package com.dc.rxjava;

import android.util.Log;

import java.util.concurrent.TimeUnit;

import io.reactivex.Notification;
import io.reactivex.Observable;
import io.reactivex.ObservableEmitter;
import io.reactivex.ObservableOnSubscribe;
import io.reactivex.ObservableSource;
import io.reactivex.Observer;
import io.reactivex.android.schedulers.AndroidSchedulers;
import io.reactivex.disposables.Disposable;
import io.reactivex.functions.Action;
import io.reactivex.functions.BooleanSupplier;
import io.reactivex.functions.Consumer;
import io.reactivex.functions.Function;
import io.reactivex.schedulers.Schedulers;

/**
 * 功能操作符
 *
 * @author 冰柠檬dc
 */
public class FunctionOperatorUtils {

    public static final String TAG = "RxJava";

    /**
     * subscribeOn操作符：指定被观察者的线程，要注意的时，如果多次调用此方法，只有第一次有效。
     *
     * @return Disposable
     */
    public static Disposable subscribeOn() {
        return Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                Log.e(TAG, "subscribe currentThread name:" + Thread.currentThread().getName());
                emitter.onNext(1);
                emitter.onNext(2);
                emitter.onComplete();
            }
        }).subscribeOn(Schedulers.io()).subscribeOn(AndroidSchedulers.mainThread()).subscribe(i -> {
            Log.e(TAG, "onNext:" + i + ",currentThread name:" + Thread.currentThread().getName());
        }, throwable -> {
            Log.e(TAG, "onError");
        }, () -> {
            Log.e(TAG, "onComplete currentThread name:" + Thread.currentThread().getName());
        }, onSubscribe -> {
            Log.e(TAG, "onSubscribe currentThread name:" + Thread.currentThread().getName());
        });
        // 打印结果：
        // onSubscribe currentThread name:main
        // subscribe currentThread name:RxCachedThreadScheduler-1
        // onNext:1,currentThread name:RxCachedThreadScheduler-1
        // onNext:2,currentThread name:RxCachedThreadScheduler-1
        // onComplete currentThread name:RxCachedThreadScheduler-1
    }

    /**
     * observeOn操作符：指定观察者的线程，每指定一次就会生效一次。
     *
     * @return Disposable
     */
    public static Disposable observeOn() {
        return Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                Log.e(TAG, "subscribe currentThread name:" + Thread.currentThread().getName());
                emitter.onNext(1);
                emitter.onNext(2);
                emitter.onComplete();
            }
        }).subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread()).map(new Function<Integer, Integer>() {
            @Override
            public Integer apply(Integer integer) throws Exception {
                Log.e(TAG, "map currentThread name:" + Thread.currentThread().getName());
                return integer + 10;
            }
        }).observeOn(Schedulers.newThread()).flatMap(new Function<Integer, ObservableSource<?>>() {
            @Override
            public ObservableSource<?> apply(Integer integer) throws Exception {
                return Observable.create(new ObservableOnSubscribe<Integer>() {
                    @Override
                    public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                        Log.e(TAG, "flatMap currentThread name:" + Thread.currentThread().getName());
                        emitter.onNext(integer);
                        emitter.onComplete();
                    }
                });
            }
        }).observeOn(Schedulers.computation()).subscribe(i -> {
            Log.e(TAG, "onNext:" + i + ",currentThread name:" + Thread.currentThread().getName());
        }, throwable -> {
            Log.e(TAG, "onError");
        }, () -> {
            Log.e(TAG, "onComplete currentThread name:" + Thread.currentThread().getName());
        }, onSubscribe -> {
            Log.e(TAG, "onSubscribe currentThread name:" + Thread.currentThread().getName());
        });
        // 打印结果：
        // onSubscribe currentThread name:main
        // subscribe currentThread name:RxCachedThreadScheduler-1
        // onNext:1,currentThread name:RxCachedThreadScheduler-1
        // onNext:2,currentThread name:RxCachedThreadScheduler-1
        // onComplete currentThread name:RxCachedThreadScheduler-1
        // onSubscribe currentThread name:main
        // subscribe currentThread name:RxCachedThreadScheduler-1
        // map currentThread name:main
        // map currentThread name:main
        // flatMap currentThread name:RxNewThreadScheduler-1
        // flatMap currentThread name:RxNewThreadScheduler-1
        // onNext:11,currentThread name:RxComputationThreadPool-1
        // onNext:12,currentThread name:RxComputationThreadPool-1
        // onComplete currentThread name:RxComputationThreadPool-1
    }

    /**
     * delay操作符：延迟一段事件发送事件。
     *
     * @return Disposable
     */
    public static Disposable delay() {
        return Observable.just(1, 2).delay(2, TimeUnit.SECONDS).subscribe(i -> {
            Log.e(TAG, "onNext:" + i);
        }, throwable -> {
            Log.e(TAG, "onError");
        }, () -> {
            Log.e(TAG, "onComplete");
        }, onSubscribe -> {
            Log.e(TAG, "onSubscribe");
        });
        // 打印结果：(打印onSubscribe等待2s后再打印剩下的事件)
        // onSubscribe
        // onNext:1
        // onNext:2
        // onComplete
    }

    /**
     * retry操作符：如果出现错误事件，则会重新发送所有事件序列。times 是代表重新发的次数。
     *
     * @return Disposable
     */
    public static Disposable retry() {
        return Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                emitter.onNext(1);
                emitter.onNext(2);
                emitter.onError(new Exception("404"));
            }
        }).retry(2).subscribe(i -> {
            Log.e(TAG, "onNext:" + i);
        }, throwable -> {
            Log.e(TAG, "onError");
        }, () -> {
            Log.e(TAG, "onComplete");
        }, onSubscribe -> {
            Log.e(TAG, "onSubscribe");
        });
        // 打印结果：
        // onSubscribe
        // onNext:1
        // onNext:2
        // onNext:1
        // onNext:2
        // onNext:1
        // onNext:2
        // onError
    }

    /**
     * retryUntil操作符：出现错误事件之后，可以通过此方法判断是否继续发送事件。
     *
     * @return Disposable
     */
    public static Disposable retryUntil() {
        return Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                emitter.onNext(1);
                emitter.onNext(2);
                emitter.onError(new Exception("404"));
            }
        }).retryUntil(new BooleanSupplier() {
            @Override
            public boolean getAsBoolean() throws Exception {
                // 停止重试
                return true;
                // 一直重试
//                return false;
            }
        }).subscribe(i -> {
            Log.e(TAG, "onNext:" + i);
        }, throwable -> {
            Log.e(TAG, "onError");
        }, () -> {
            Log.e(TAG, "onComplete");
        }, onSubscribe -> {
            Log.e(TAG, "onSubscribe");
        });
        // 打印结果：
        // onSubscribe
        // onNext:1
        // onNext:2
        // onError
    }

    /**
     * retryWhen操作符：当被观察者接收到异常或者错误事件时会回调该方法，这个方法会返回一个新的被观察者。
     * 如果返回的被观察者发送 Error 事件则之前的被观察者不会继续发送事件，如果发送正常事件则之前的被观察者会继续不断重试发送事件。
     *
     * @return Disposable
     */
    public static Disposable retryWhen() {
        return Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                emitter.onNext(1);
                emitter.onNext(2);
                emitter.onError(new Exception("404"));
                emitter.onNext(3);
            }
        }).retryWhen(new Function<Observable<Throwable>, ObservableSource<?>>() {
            @Override
            public ObservableSource<?> apply(Observable<Throwable> throwableObservable) throws Exception {
                return throwableObservable.flatMap(new Function<Throwable, ObservableSource<?>>() {
                    @Override
                    public ObservableSource<?> apply(Throwable throwable) throws Exception {
                        if (!throwable.toString().equals("java.lang.Exception: 404")) {
                            return Observable.just("可以忽略的异常");
                        } else {
                            return Observable.error(new Throwable("终止啦"));
                        }
                    }
                });
            }
        }).subscribe(i -> {
            Log.e(TAG, "onNext:" + i);
        }, throwable -> {
            Log.e(TAG, "onError:" + throwable);
        }, () -> {
            Log.e(TAG, "onComplete");
        }, onSubscribe -> {
            Log.e(TAG, "onSubscribe");
        });
        // 打印结果：
        // onSubscribe
        // onNext:1
        // onNext:2
        // onError:java.lang.Throwable: 终止啦
    }

    /**
     * repeat操作符：重复发送被观察者的事件，times 为发送次数。
     *
     * @return Disposable
     */
    public static Disposable repeat() {
        return Observable.just(1, 2).repeat(2).subscribe(i -> {
            Log.e(TAG, "onNext:" + i);
        }, throwable -> {
            Log.e(TAG, "onError");
        }, () -> {
            Log.e(TAG, "onComplete");
        }, onSubscribe -> {
            Log.e(TAG, "onSubscribe");
        });
        // 打印结果：
        // onSubscribe
        // onNext:1
        // onNext:2
        // onNext:1
        // onNext:2
        // onComplete
    }

    /**
     * repeatUntil操作符：可以通过此方法判断是否继续发送事件。
     *
     * @return Disposable
     */
    public static Disposable repeatUntil() {
        return Observable.just(1, 2).repeatUntil(new BooleanSupplier() {
            @Override
            public boolean getAsBoolean() throws Exception {
                // 停止重复发送
                return true;
                // 一直重复发送
//                return false;
            }
        }).subscribe(i -> {
            Log.e(TAG, "onNext:" + i);
        }, throwable -> {
            Log.e(TAG, "onError");
        }, () -> {
            Log.e(TAG, "onComplete");
        }, onSubscribe -> {
            Log.e(TAG, "onSubscribe");
        });
        // 打印结果：
        // onSubscribe
        // onNext:1
        // onNext:2
        // onComplete
    }

    /**
     * repeatWhen操作符：返回一个新的被观察者设定一定逻辑来决定是否重复发送事件。
     *
     * @return Disposable
     */
    public static Disposable repeatWhen() {
        return Observable.just(1, 2).repeatWhen(new Function<Observable<Object>, ObservableSource<?>>() {
            @Override
            public ObservableSource<?> apply(Observable<Object> objectObservable) throws Exception {
//                return Observable.empty();
//                return Observable.error(new Exception("404"));
                return Observable.timer(3, TimeUnit.SECONDS);
            }
        }).subscribe(i -> {
            Log.e(TAG, "onNext:" + i);
        }, throwable -> {
            Log.e(TAG, "onError");
        }, () -> {
            Log.e(TAG, "onComplete");
        }, onSubscribe -> {
            Log.e(TAG, "onSubscribe");
        });
        // 打印结果：(发送其他事件，3s后重复一次)
        // onSubscribe
        // onNext:1
        // onNext:2
        // onNext:1
        // onNext:2
        // onComplete

        // 打印结果：(发送empty)
        // onSubscribe
        // onComplete

        // 打印结果：(发送error)
        // onSubscribe
        // onError
    }

    /**
     * doOnEach操作符：Observable 每发送一件事件之前都会先回调这个方法。
     *
     * @return Disposable
     */
    public static Disposable doOnEach() {
        return Observable.just(1, 2).doOnEach(new Consumer<Notification<Integer>>() {
            @Override
            public void accept(Notification<Integer> integerNotification) throws Exception {
                Log.e(TAG, "doOnEach:" + integerNotification.getValue());
            }
        }).subscribe(i -> {
            Log.e(TAG, "onNext:" + i);
        }, throwable -> {
            Log.e(TAG, "onError");
        }, () -> {
            Log.e(TAG, "onComplete");
        }, onSubscribe -> {
            Log.e(TAG, "onSubscribe");
        });
        // 打印结果：
        // onSubscribe
        // doOnEach:1
        // onNext:1
        // doOnEach:2
        // onNext:2
        // doOnEach:null
        // onComplete
    }

    /**
     * doOnNext操作符：Observable 每发送 onNext() 之前都会先回调这个方法。
     *
     * @return Disposable
     */
    public static Disposable doOnNext() {
        return Observable.just(1, 2).doOnNext(new Consumer<Integer>() {
            @Override
            public void accept(Integer integer) throws Exception {
                Log.e(TAG, "doOnNext:" + integer);
            }
        }).subscribe(i -> {
            Log.e(TAG, "onNext:" + i);
        }, throwable -> {
            Log.e(TAG, "onError");
        }, () -> {
            Log.e(TAG, "onComplete");
        }, onSubscribe -> {
            Log.e(TAG, "onSubscribe");
        });
        // 打印结果：(发送最后一个事件后不会发送null)
        // onSubscribe
        // doOnNext:1
        // onNext:1
        // doOnNext:2
        // onNext:2
        // onComplete
    }

    /**
     * doAfterNext操作符：OObservable 每发送 onNext() 之后都会回调这个方法。
     *
     * @return Disposable
     */
    public static Disposable doAfterNext() {
        return Observable.just(1, 2).doAfterNext(new Consumer<Integer>() {
            @Override
            public void accept(Integer integer) throws Exception {
                Log.e(TAG, "doAfterNext:" + integer);
            }
        }).subscribe(i -> {
            Log.e(TAG, "onNext:" + i);
        }, throwable -> {
            Log.e(TAG, "onError");
        }, () -> {
            Log.e(TAG, "onComplete");
        }, onSubscribe -> {
            Log.e(TAG, "onSubscribe");
        });
        // 打印结果：
        // onSubscribe
        // onNext:1
        // doAfterNext:1
        // onNext:2
        // doAfterNext:2
        // onComplete
    }

    /**
     * doOnSubscribe操作符：Observable 每发送 onSubscribe() 之前都会回调这个方法。
     * doOnComplete操作符：Observable 每发送 onComplete() 之前都会回调这个方法。
     * doOnError操作符：Observable 每发送 onError() 之前都会回调这个方法。
     *
     * @return Disposable
     */
    public static Disposable doOnSubscribe() {
        return Observable.just(1, 2).doOnSubscribe(new Consumer<Disposable>() {
            @Override
            public void accept(Disposable disposable) throws Exception {
                Log.e(TAG, "doOnSubscribe");
            }
        }).doOnComplete(new Action() {
            @Override
            public void run() throws Exception {
                Log.e(TAG, "doOnComplete");
            }
        }).doOnError(new Consumer<Throwable>() {
            @Override
            public void accept(Throwable throwable) throws Exception {
                Log.e(TAG, "doOnError");
            }
        }).subscribe(i -> {
            Log.e(TAG, "onNext:" + i);
        }, throwable -> {
            Log.e(TAG, "onError");
        }, () -> {
            Log.e(TAG, "onComplete");
        }, onSubscribe -> {
            Log.e(TAG, "onSubscribe");
        });
        // 打印结果：
        // onSubscribe
        // doOnSubscribe
        // onNext:1
        // onNext:2
        // doOnComplete
        // onComplete
    }

    /**
     * doOnDispose操作符：当调用 Disposable 的 dispose() 之后回调该方法。
     */
    public static void doOnDispose() {
        Observable.just(1, 2).doOnDispose(new Action() {
            @Override
            public void run() throws Exception {
                Log.e(TAG, "doOnDispose");
            }
        }).subscribe(new Observer<Integer>() {
            private Disposable d;

            @Override
            public void onSubscribe(Disposable d) {
                Log.e(TAG, "onSubscribe");
                this.d = d;
            }

            @Override
            public void onNext(Integer integer) {
                Log.e(TAG, "onNext: " + integer);
                d.dispose();
            }

            @Override
            public void onError(Throwable e) {
                Log.e(TAG, "onError");
            }

            @Override
            public void onComplete() {
                Log.e(TAG, "onComplete");
            }
        });
        // 打印结果：
        // onSubscribe
        // onNext:1
        // doOnDispose
    }

    /**
     * doOnLifecycle操作符：在回调 onSubscribe 之前回调该方法的第一个参数的回调方法，可以使用该回调方法决定是否取消订阅。
     */
    public static void doOnLifecycle() {
        Observable.just(1, 2).doOnLifecycle(new Consumer<Disposable>() {
            @Override
            public void accept(Disposable disposable) throws Exception {
                Log.e(TAG, "doOnLifecycle accept");
                disposable.dispose();
            }
        }, new Action() {
            @Override
            public void run() throws Exception {
                Log.e(TAG, "doOnLifecycle run");
            }
        }).doOnDispose(new Action() {
            @Override
            public void run() throws Exception {
                Log.e(TAG, "doOnDispose");
            }
        }).subscribe(new Observer<Integer>() {
            private Disposable d;

            @Override
            public void onSubscribe(Disposable d) {
                Log.e(TAG, "onSubscribe");
                this.d = d;
            }

            @Override
            public void onNext(Integer integer) {
                Log.e(TAG, "onNext: " + integer);
//                d.dispose();
            }

            @Override
            public void onError(Throwable e) {
                Log.e(TAG, "onError");
            }

            @Override
            public void onComplete() {
                Log.e(TAG, "onComplete");
            }
        });
        // 打印结果：(onNext() 方法进行取消订阅操作后，doOnDispose() 和 doOnLifecycle run 都会被回调。)
        // doOnLifecycle accept
        // onSubscribe
        // onNext: 1
        // doOnDispose
        // doOnLifecycle run

        // 打印结果：(doOnLifecycle 进行取消订阅，doOnDispose 和 doOnLifecycle  run都没有被回调。)
        // doOnLifecycle accept
        // onSubscribe
    }

    /**
     * doOnTerminate操作符：在 onError 或者 onComplete 发送之前回调
     * doAfterTerminate操作符：在 onError 或者 onComplete 发送之后回调
     *
     * @return Disposable
     */
    public static Disposable doOnTerminate() {
        return Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                emitter.onNext(1);
                emitter.onNext(2);
//                emitter.onError(new NullPointerException());
                emitter.onComplete();
            }
        }).subscribe(i -> {
            Log.e(TAG, "onNext:" + i);
        }, throwable -> {
            Log.e(TAG, "onError");
        }, () -> {
            Log.e(TAG, "onComplete");
        }, onSubscribe -> {
            Log.e(TAG, "onSubscribe");
        });
        // 打印结果：
        // onSubscribe
        // onNext:1
        // onNext:2
        // doOnTerminate
        // onComplete
    }

    /**
     * doOnTerminate操作符：在所有事件发送完毕之后回调该方法。
     * doFinally() 和 doAfterTerminate() 到底有什么区别？区别就是在于取消订阅，、
     * 如果取消订阅之后 doAfterTerminate() 就不会被回调，而 doFinally() 无论怎么样都会被回调，且都会在事件序列的最后。
     */
    public static void doFinally() {
        Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                emitter.onNext(1);
                emitter.onNext(2);
//                emitter.onError(new NullPointerException());
                emitter.onComplete();
            }
        }).doFinally(new Action() {
            @Override
            public void run() throws Exception {
                Log.e(TAG, "doFinally");
            }
        }).doOnDispose(new Action() {
            @Override
            public void run() throws Exception {
                Log.e(TAG, "doOnDispose");
            }
        }).doAfterTerminate(new Action() {
            @Override
            public void run() throws Exception {
                Log.e(TAG, "doAfterTerminate");
            }
        }).subscribe(new Observer<Integer>() {
            private Disposable d;
            @Override
            public void onSubscribe(Disposable d) {
                Log.e(TAG, "onSubscribe");
                this.d = d;
            }

            @Override
            public void onNext(Integer integer) {
                Log.e(TAG, "onNext:" + integer);
//                d.dispose();
            }

            @Override
            public void onError(Throwable e) {
                Log.e(TAG, "onError");
            }

            @Override
            public void onComplete() {
                Log.e(TAG, "onComplete");
            }
        });
        // 打印结果：(调用了 dispose() 方法，doAfterTerminate() 不会被回调。)
        // onSubscribe
        // onNext:1
        // doOnDispose
        // doFinally

        // 打印结果：(不调用 dispose() 方法，doAfterTerminate() 会被回调。)
        // onSubscribe
        // onNext:1
        // onNext:2
        // onComplete
        // doAfterTerminate
        // doFinally
    }

    /**
     * onErrorReturn操作符：接受到一个 onError() 事件之后回调，返回的值会回调 onNext() 方法，并正常结束该事件序列。
     *
     * @return Disposable
     */
    public static Disposable onErrorReturn() {
        return Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                emitter.onNext(1);
                emitter.onNext(2);
                emitter.onError(new NullPointerException());
            }
        }).onErrorReturn(new Function<Throwable, Integer>() {
            @Override
            public Integer apply(Throwable throwable) throws Exception {
                Log.e(TAG, "onErrorReturn:" + throwable);
                return 404;
            }
        }).subscribe(i -> {
            Log.e(TAG, "onNext:" + i);
        }, throwable -> {
            Log.e(TAG, "onError");
        }, () -> {
            Log.e(TAG, "onComplete");
        }, onSubscribe -> {
            Log.e(TAG, "onSubscribe");
        });
        // 打印结果：
        // onSubscribe
        // onNext:1
        // onNext:2
        // onErrorReturn:java.lang.NullPointerException
        // onNext:404
        // onComplete
    }

    /**
     * onErrorResumeNext操作符：当接收到 onError() 事件时，返回一个新的 Observable，并正常结束事件序列。
     *
     * @return Disposable
     */
    public static Disposable onErrorResumeNext() {
        return Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                emitter.onNext(1);
                emitter.onNext(2);
                emitter.onError(new NullPointerException());
            }
        }).onErrorResumeNext(new Function<Throwable, ObservableSource<? extends Integer>>() {
            @Override
            public ObservableSource<? extends Integer> apply(Throwable throwable) throws Exception {
                Log.e(TAG, "onErrorResumeNext:" + throwable);
                return Observable.just(3, 4);
            }
        }).subscribe(i -> {
            Log.e(TAG, "onNext:" + i);
        }, throwable -> {
            Log.e(TAG, "onError");
        }, () -> {
            Log.e(TAG, "onComplete");
        }, onSubscribe -> {
            Log.e(TAG, "onSubscribe");
        });
        // 打印结果：
        // onSubscribe
        // onNext:1
        // onNext:2
        // onErrorReturn:java.lang.NullPointerException
        // onNext:3
        // onNext:4
        // onComplete
    }

    /**
     * onExceptionResumeNext操作符：与 onErrorResumeNext() 作用基本一致，但是这个方法只能捕捉 Exception。
     *
     * @return Disposable
     */
    public static Disposable onExceptionResumeNext() {
        return Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                emitter.onNext(1);
                emitter.onNext(2);
//                emitter.onError(new Error("404"));
                emitter.onError(new Exception("404"));
            }
        }).onExceptionResumeNext(new ObservableSource<Integer>() {
            @Override
            public void subscribe(Observer<? super Integer> observer) {
                observer.onNext(3);
                observer.onComplete();
            }
        }).subscribe(i -> {
            Log.e(TAG, "onNext:" + i);
        }, throwable -> {
            Log.e(TAG, "onError");
        }, () -> {
            Log.e(TAG, "onComplete");
        }, onSubscribe -> {
            Log.e(TAG, "onSubscribe");
        });
        // 打印结果：(发送Error)
        // onSubscribe
        // onNext:1
        // onNext:2
        // onError

        // 打印结果：(发送Exception)
        // onSubscribe
        // onNext:1
        // onNext:2
        // onNext:3
        // onComplete
    }
}
